package com.zx.learn.flink.apps.app1mysqltomysql;

import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;

/**
 * MySQL 到MySQL
 */
@Slf4j
public class MySQLToMySQLMainKeyBy {
    public static void main(String[] args) throws Exception {
        //1.env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //2.source
        DataStreamSource<SampleUser> sourceDS = env.addSource(new MySQLSource<SampleUser>() {
            @Override
            public void run(SourceContext<SampleUser> sourceContext) throws Exception {
                ResultSet resultSet = ps.executeQuery();
                while (resultSet.next()) {
                    SampleUser obj = new SampleUser();
                    obj.setId(resultSet.getInt("id"));
                    obj.setName(resultSet.getString("name"));
                    obj.setAge(resultSet.getInt("age"));
                    sourceContext.collect(obj);
                }
            }

            @Override
            Connection getConnection() {
                Connection con = null;
                try {
                    Class.forName("com.mysql.cj.jdbc.Driver");
                    //注意，替换成自己本地的 mysql 数据库地址和用户名、密码
                    con = DriverManager.getConnection("jdbc:mysql://localhost:3306/bigdata?characterEncoding=utf8&useSSL=false&serverTimezone=UTC&rewriteBatchedStatements=true&autoReconnect=true", "root", "root");
                } catch (Exception e) {
                    log.error("get connection has exception , msg = {}", e.getMessage());
                }
                return con;
            }

            @Override
            String getSql() {
                return "select * from sample_user limit 100";
            }

        });

        sourceDS.print();

        //transfer
        MapFunction map = new MapFunction<SampleUser, SampleUser>() {
            @Override
            public SampleUser map(SampleUser value) throws Exception {
                value.setAge(value.getAge() + 1);
                return value;
            }
        };
        SingleOutputStreamOperator<SampleUser> mapResult = sourceDS.map(map);
        mapResult.print(" 1、transfer:map 一年后:");
        KeyedStream<SampleUser, Integer> keyByStreamObj = mapResult.keyBy(new KeySelector<SampleUser, Integer>() {

            @Override
            public Integer getKey(SampleUser value) throws Exception {
                return value.getAge();
            }
        });
        keyByStreamObj.print(" 1.1、transfer:map:keyByStreamObj -- keyby object：");

        //lamada
        map = (MapFunction<SampleUser, SampleUser>) value -> {
            value.setAge(value.getAge() + 1);
            return value;
        };
        mapResult = sourceDS.map(map);
        mapResult.print(" 2、transfer:map:lamada 一年后:");
        keyByStreamObj = mapResult.keyBy((KeySelector<SampleUser, Integer>) value -> value.getAge());
        keyByStreamObj.print(" 2.1、transfer:map:lamada:keyByStreamObj -- keyby lamada：");

        SingleOutputStreamOperator<Tuple2<String, Integer>> flatResult = sourceDS.flatMap((FlatMapFunction<SampleUser, Tuple2<String, Integer>>) (value, out) -> {
            out.collect(Tuple2.of("id:" + value.getId(), 1));
            out.collect(Tuple2.of("age:" + value.getAge(), 1));
        }).returns(Types.TUPLE(Types.STRING, Types.INT));
        flatResult.print(" 3、transfer:flatMap :");
        KeyedStream<Tuple2<String, Integer>, String> keyByStreamTuple = flatResult.keyBy((KeySelector<Tuple2<String, Integer>, String>) t -> t.f0);
        keyByStreamTuple.print(" 3.1、transfer:map:lamada:keyByStreamTuple -- keyby tuple2：");

        SingleOutputStreamOperator<String> flatResultString = sourceDS.flatMap((FlatMapFunction<SampleUser, String>) (value, out) -> {
            out.collect(value.getName());
            out.collect("id:" + value.getId());
            out.collect("age:" + value.getAge());
        }).returns(String.class);
        flatResultString.print(" 4、transfer:flatMap String :");
        KeyedStream<String , String> keyByStreamStr = flatResultString.keyBy((KeySelector<String, String>) t -> t);
        keyByStreamStr.print(" 4.1、transfer:map:lamada:keyByStreamStr -- keyby str：");

        SingleOutputStreamOperator<SampleUser> flatResultObj = sourceDS.flatMap((FlatMapFunction<SampleUser, SampleUser>) (value, out) -> {
            value.setName(value.getName() + "_" + value.getAge() + "_" + value.getId());
            value.setId(0);
            value.setAge(1);
            out.collect(value);
        }).returns(SampleUser.class);
        flatResultObj.print(" 5、transfer:flatMapObj :");

        KeyedStream<SampleUser , Integer> keyByStreamObject = flatResultObj.keyBy((KeySelector<SampleUser, Integer>) SampleUser::getAge);
        keyByStreamObject.print(" 5.1、transfer:map:lamada:keyByStreamStr -- keyby Object：");

        
        env.execute();
    }
}
